package com.telra.app.server.cassandra;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.telra.app.server.data.OrderEvent;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;


import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

public class CassandraWriter {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(CassandraWriter.class);

    private Cluster cluster;
    private Keyspace keyspaceOperator;
    private BlockingQueue<OrderEvent> queue;
    private static StringSerializer stringSerializer = StringSerializer.get();

    public CassandraWriter() {

        queue = new LinkedBlockingQueue<OrderEvent>();

        //get the cluster
        cluster = HFactory.getOrCreateCluster("TestCluster", "localhost:9160");
        //get the keyspace
        keyspaceOperator = HFactory.createKeyspace("OrderKeyspace", cluster);

        Thread t = new Thread(new Insertor());
        t.setName("cassandraInsert");
        t.start();
    }

    public void queueInsert(OrderEvent order) {
        try {
            queue.put(order);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private class Insertor implements Runnable {

        public void run() {

            while (true) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                List<OrderEvent> orders = new ArrayList<OrderEvent>();
                queue.drainTo(orders);

                log.info("queue:" + orders.size());

                Mutator<String> mutator = HFactory.createMutator(keyspaceOperator, stringSerializer);

                int c= 0;
                for (OrderEvent order : orders) {
                    //insert(key, colFamily, column<key,val>)
                    mutator.
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("itemName", order.getItemName())).
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("price", String.valueOf(order.getPrice()))).
                            addInsertion(order.getKey(), "Orders",
                                    HFactory.createStringColumn("category", order.getCategory().toString()));

                    if (c++ > 10000) {
                        mutator.execute();
                        c = 0;
                    }
                }

                if (c > 0) {	//flush the rest!
                    mutator.execute();
                }

                log.info("queue processed");
            }
        }

    }
}

